
/*
Copyright 2009 Peter Ekler, Akos Ludanyi, Budapest University of Technology and Economics, Department of Automation and Applied Informatics, Applied Mobile Research Group (AMORG) http://amorg.aut.bme.hu

This file is part of MobTorrent.

MobTorrent is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

MobTorrent is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with MobTorrent.  If not, see <http://www.gnu.org/licenses/>.
*/

/*
 * MTPeerConnection.java
 *
 * Created on 2007. m�rcius 19., 12:43
 *
 * To change this template, choose Tools | Template Manager
 * and open the template in the editor.
 */
package TorrentEngine;

import Logger.MTLogger;
import Settings.MTErrorCodes;
import Settings.Preferences;
import Tools.NetTools;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.util.Vector;

import javax.microedition.io.Connector;
import javax.microedition.io.SocketConnection;

/**
 *
 * @author Tyrial
 */
public class MTPeerConnection 
{

    // ---Consts
    public final static int KMaxPieceRequests = 2;
    private final String ProtocolId = "BitTorrent protocol";
    public final static int KMessageIdChoke = 0;
    public final static int KMessageIdUnchoke = 1;
    public final static int KMessageIdInterested = 2;
    public final static int KMessageIdNotInterested = 3;
    public final static int KMessageIdHave = 4;
    public final static int KMessageIdBitfield = 5;
    public final static int KMessageIdRequest = 6;
    public final static int KMessageIdPiece = 7;
    public final static int KMessageIdCancel = 8;
    public final static int KTcpConnectTimeout = 15;
    public final static int KHandshakeTimeout = 15;
    public final static int KPwConnectionTimeout = 2 * 60;
    public final static int KKeepAliveInterval = 2 * 60; // once in every two minutes
    public final static int KRequestTimeout = 60;
    private final static int KDefaultBlockLength = 16384; // default download block size (2^14)    
    private final static int KFlagAmChoking = 1; 	// ...0001
    private final static int KFlagAmInterested = 2;	// ...0010
    private final static int KFlagPeerChoking = 4; 	// ...0100
    private final static int KFlagPeerInterested = 8; 	// ...1000
    // ---Enums
    public final static int EPeerNotConnected = 0;
    public final static int EPeerTcpConnecting = 1;
    public final static int EPeerConnected = 2;
    public final static int EPeerPwHandshaking = 3;
    public final static int EPeerPwConnected = 4;
    public final static int EPeerClosing = 5;
    public final static int EDeletePeer = 0;
    public final static int EIncreaseErrorCounter = 1;
    public final static int ENotSpecified = 2;
    // ---Variables
    private MTPeer peer;
    private MTTorrent torrent;
    private MTTorrentManager torrentMgr;
    private boolean incomingConnection;
    private int retries;
    private int reconnectAfter;
    private int state;
    private int ellapsedTime;
    private int lastRequestTime;
    private int lastMessageReceivedTime;
    private int lastMessageSentTime;
    private int closeOrder;
    private boolean peerWireConnected = false;
    final private Vector piecesToDownload;
    private int statusFlags;
    private boolean hasPendingDownloadRequest;
    public Vector incomingRequests;
    // ---Network variables    
    private SocketConnection socket;
    private InputStream inputStream;
    private OutputStream outputStream;
    private boolean readEnabled;
    private ConnectThread connectThread = null;
    public static int tcpConnectionTimeoutNum = 0;
    public static boolean useLongConnection = false;

    private PeerSenderInterface peerSender;


    // Store class for incomming requests
    class BlockRequest
    {

        public int pieceIndex;
        public int begin;
        public int length;

        public BlockRequest(int aPieceIndex, int aBegin, int aLength)
        {
            pieceIndex = aPieceIndex;
            begin = aBegin;
            length = aLength;
        }
    };

    private boolean readData(byte [] data) throws Exception
    {
        return readData(data, 0, data.length);
    }

    private boolean readData(byte [] data, int offset, int len) throws Exception
    {
        int remain = len;
        while(remain > 0)
        {
            final int readed = inputStream.read(data, offset, remain);
            if(readed == -1)
            {
                break;
            }
            remain -= readed;
            offset += readed;
        }

        return remain == 0;
    }

    /**
     * This class is a Thread, which reads the messages from the connection
     */
    private void read()
    {
        int messageLength = 0;

        while(readEnabled)
        {
            try
            {
                switch(state)
                {
                    case EPeerPwHandshaking:
                    {
                        int protLength = inputStream.read();

                        if(protLength == -1) // -1 error from the stream: connection closed
                        {
                            close(EIncreaseErrorCounter, "Peer disconnected!");
                            continue;
                        }

                        int handshakeLength = protLength + 48; // -1 because of protLength

                        byte[] handshake = new byte[handshakeLength];
                        //inputStream.read(handshake);
                        readData(handshake);

                        byte[] otherProtocolId = new byte[protLength];
                        System.arraycopy(handshake, 0, otherProtocolId, 0, protLength);

                        if(!ProtocolId.equals(new String(otherProtocolId)))
                        {
                            close(EDeletePeer, "Protocol identifier doesn't match!");
                            continue;
                        }

                        byte[] infoHash = new byte[20];
                        System.arraycopy(handshake, 27, infoHash, 0, 20);

                        if(torrent != null)
                        {
                            if(!NetTools.byteArrayEqual(infoHash, torrent.getInfoHashByteArray()))
                            {
                                close(EDeletePeer, "Torrent infohash doesn't match!");
                                continue;
                            }
                        }
                        else // if torrent is null then we should attach peer to torrent (most likely it is an incoming connection)
                        {
                            if((torrentMgr.attachPeerToTorrent(infoHash, getPeerConnection()) != MTErrorCodes.ErrNone) || (torrent == null)) // close if the attach failed
                            {
                                close(EDeletePeer, "Invalid infohash or peer is already connected or too many peers!");
                                continue;
                            }
                        }

                        if(incomingConnection)
                        {
                            torrentMgr.notifyTorrentObserver(torrent, MTTorrentObserver.EMTEventIncomingConnectionsChanged);
                            torrent.incIncomingConnectionsNum();
                            sendHandshakeMessage();
                            peer.resetAddress();
                        }

                        byte[] peerId = new byte[20];
                        System.arraycopy(handshake, 47, peerId, 0, 20);
                        String tempPId = new String(peerId);
                        if(peer.getPeerId() != null)
                        {
                            if(!peer.getPeerId().equals(tempPId))
                            {
                                close(EIncreaseErrorCounter, "Peer ID doesn't match!");
                                continue;
                            }
                            else if(tempPId.equals(torrentMgr.getPeerID()))
                            {
                                close(EDeletePeer, "Connected to ourselves!");
                                continue;
                            }
                        }
                        else
                        {
                            peer.setPeerId(tempPId);
                        }

                        log("Handshake completed! Peer wire connected!");
                        changeState(MTPeerConnection.EPeerPwConnected);
                        setPeerWireConnected();

                        if(!torrent.getBitField().isNull())
                            sendBitfieldMessage();

                        break;
                    }
                    case EPeerPwConnected:
                    {
                        byte[] initData = new byte[4];

                        //inputStream.read(initData); // BLOCKS
                        readData(initData);

                        messageLength = getInt(initData);

                        lastMessageReceivedTime = ellapsedTime;
                        if(messageLength == 0)
                        {
                            //log("in KEEPALIVE");
                            //TODOAKOS: we have to close connection,
                            issueDownload();
                        }
                        else
                        {
                            int id = inputStream.read();
                            switch(id)
                            {
                                case KMessageIdBitfield:
                                {
                                    log("in BITFIELD");
                                    if(messageLength - 1 != peer.getBitField().lengthInBytes())
                                    {
                                        close(EIncreaseErrorCounter, "Received bitfield length doesn't match!");
                                    }
                                    else
                                    {
                                        int bitFieldLength = messageLength - 1; // -id length
                                        byte[] bitFieldDes = new byte[bitFieldLength];
                                        //int dataRead = inputStream.read(bitFieldDes);
                                        boolean dataReaded = readData(bitFieldDes);

                                        //String al = new String(bitFieldDes);
                                        if(dataReaded)
                                        {
                                            peer.havePieces(bitFieldDes, torrent);
                                            issueDownload();
                                        }
                                        else
                                        {
                                            close(EIncreaseErrorCounter, "Could not read bitfield!");
                                        }
                                    }
                                    break;
                                }

                                case KMessageIdHave:
                                    {
                                        log("in HAVE");
                                        int pieceIndex = readInt(0);

                                        if((pieceIndex >= 0) && (pieceIndex < torrent.pieceCount()))
                                        {
                                            peer.havePiece(pieceIndex, torrent);

                                            if(!hasPendingDownloadRequest)
                                                issueDownload();
                                        }

                                    }
                                    break;

                                case KMessageIdChoke:
                                {
                                    peer.resetErrorCounter();
                                    log("in CHOKE");
                                    setPeerChoking(true);

                                    synchronized(piecesToDownload)
                                    {
                                        for(int i = 0; i < piecesToDownload.size(); i++)
                                            ((MTPieceToDownload) piecesToDownload.elementAt(i)).hasPendingRequest = false;

                                        for(int i = 0; i < piecesToDownload.size(); i++)
                                            torrent.removePieceFromDownloading(((MTPieceToDownload) piecesToDownload.elementAt(i)).piece);

                                        piecesToDownload.removeAllElements();
                                    }

                                    break;
                                }

                                case KMessageIdUnchoke:
                                {
                                    peer.resetErrorCounter();
                                    log("in UNCHOKE");
                                    setPeerChoking(false);
                                    issueDownload();
                                    break;
                                }

                                case KMessageIdInterested:
                                {
                                    peer.resetErrorCounter();
                                    log("in INTERESTED");
                                    if(Preferences.UploadEnabled)
                                    {
                                        setPeerInterested(true);
                                        setChoking(false);
                                    }
                                    MTLogger.writeLine("--- INTERESTED ARRIVED ---");
                                    break;
                                }

                                case KMessageIdNotInterested:
                                {
                                    peer.resetErrorCounter();
                                    log("in NOTINTERESTED");
                                    setPeerInterested(false);
                                    issueDownload();
                                    break;
                                }

                                case KMessageIdPiece:
                                {
                                    peer.resetErrorCounter();
                                    lastRequestTime = 0;
                                    peer.setHadRequestTimeout(false);
                                    int index = readInt(0);
                                    int begin = readInt(0);
                                    int pieceBlockSize = messageLength - 9;

                                    log("in PIECE Index: " + index + " Begin: " + begin + " Length: " + pieceBlockSize);

                                    MTPieceToDownload piece = null;
                                    synchronized(piecesToDownload)
                                    {
                                        for(int i = 0; i < piecesToDownload.size(); i++)
                                            if(((MTPieceToDownload) piecesToDownload.elementAt(i)).piece.index() == index)
                                            {
                                                piece = (MTPieceToDownload) piecesToDownload.elementAt(i);
                                                break;
                                            }
                                    }

                                    if(piece == null)
                                    {
                                        close("Error, unexpected piece (there are no pending request for the received piece index)");
                                    }
                                    else
                                    {
                                        // create PIECE byte array
                                        MTLogger.writeMemoryInfo();
                                        byte[] pieceBlock = new byte[pieceBlockSize];
                                        boolean successfullRead = readData(pieceBlock);

                                        if(!successfullRead)
                                        {
                                            close(EIncreaseErrorCounter, "Reading piece failed!");
                                            return;
                                        }

                                        /*while(length != 0)
                                        {
                                            readed = inputStream.read(pieceBlock, off, length);
                                            length = length - readed;
                                            off = off + readed;
                                        }*/

                                        piece.hasPendingRequest = false;
                      

                                        /*if(torrent.isEndGame())
                                            torrent.endGamePieceReceived(piece.piece, peer);*/

                                        if(piece.piece.appendBlock(pieceBlock, begin, peer) != MTErrorCodes.ErrNone)
                                        {
                                            pieceBlock = null;
                                            close("Writing to piece failed"); // CRITICAL FAULT
                                        }
                                        else
                                        {
                                            pieceBlock = null;
                                            /// WHY??? (piece->Remaining() == piece->TotalSize())
                                            if((piece.piece.remaining() == 0) || (piece.piece.remaining() == piece.piece.getTotalSize()))
                                            {
                                                synchronized(piecesToDownload)
                                                {
                                                    piecesToDownload.removeElement(piece);
                                                }
                                            }

                                        }
                                    }
                                    System.gc();

                                    issueDownload();

                                    break;
                                }

                                case KMessageIdRequest:
                                {
                                    peer.resetErrorCounter();
                                    if(messageLength < 13)
                                    {
                                        close(EIncreaseErrorCounter, "Received request message length is smaller than 13!");
                                    }
                                    else
                                    {
                                        int pieceIndex = readInt(5);
                                        int begin = readInt(9);
                                        int length = readInt(13);

                                        log("in REQUEST Index: " + pieceIndex + " Begin: " + begin + " Length: " + length);

                                        if(Preferences.UploadEnabled)
                                        {
                                            ellapsedTime = 0; /////////////////////////////////////////////

                                            incomingRequests.addElement(new BlockRequest(pieceIndex, begin, length));

                                            issueUpload();
                                        }
                                    }
                                    MTLogger.writeLine("--- REQUEST ARRIVED ---");
                                    break;
                                }

                                case KMessageIdCancel:
                                    {
                                        log("in CANCEL");

                                        peer.resetErrorCounter();
                                        if(messageLength < 13)
                                        {
                                            close(EIncreaseErrorCounter, "Received CANCEL message length is smaller than 13!");
                                        }
                                        else
                                        {
                                            int pieceIndex = readInt(5);
                                            int begin = readInt(9);
                                            int length = readInt(13);

                                            for(int i = 0; i < incomingRequests.size(); i++)
                                            {
                                                if(((BlockRequest) incomingRequests.elementAt(i)).pieceIndex == pieceIndex &&
                                                    ((BlockRequest) incomingRequests.elementAt(i)).begin == begin &&
                                                    ((BlockRequest) incomingRequests.elementAt(i)).length == length)
                                                {
                                                    incomingRequests.removeElementAt(i);
                                                    break;
                                                }
                                            }
                                        }
                                    }
                                    break;

                            }
                        }

                        break;
                    }
                }
            }
            catch(InterruptedIOException e)
            {
                close(EIncreaseErrorCounter, "Read error");
                MTLogger.writeLine("--- PeerConnection interrupted exception: " + e.getMessage());
            }
            catch(IOException e)
            {
                e.printStackTrace();
                close(EIncreaseErrorCounter, "Read error");
                MTLogger.writeLine("--- PeerConnection ioexception: " + e.getMessage());
            }
            catch(Exception e)
            {
                e.printStackTrace();
                close(EIncreaseErrorCounter, "Read error - CHECKOLNI!!!");
                MTLogger.writeLine("[Read Exception] " + e.getMessage());
            }
        }
    }

    public int readInt(int aIndex)
    {
        //TPtrC8 ptr = iRecvBuffer->Ptr().Right(
        //        iRecvBuffer->Ptr().Length() - aIndex);        

        byte[] in = new byte[4];
        try
        {
            //inputStream.read(in);
            readData(in);
        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Read error");
            ex.printStackTrace();
            return -1;
        }
        catch(Exception ex)
        {
            close(EIncreaseErrorCounter, "Read error");
            ex.printStackTrace();
            return -1;
        }

        int value;

        if(in[0] < 0)
            value = (256 + in[0] << 24);
        else
            value = (in[0] << 24);

        if(in[1] < 0)
            value += (256 + in[1] << 16);
        else
            value += (in[1] << 16);

        if(in[2] < 0)
            value += (256 + in[2] << 8);
        else
            value += (in[2] << 8);

        if(in[3] < 0)
            value += (256 + in[3]);
        else
            value += in[3];

        return value;
    }

    public int getInt(byte[] aIn)
    {
        int value;

        if(aIn[0] < 0)
            value = (256 + aIn[0] << 24);
        else
            value = (aIn[0] << 24);

        if(aIn[1] < 0)
            value += (256 + aIn[1] << 16);
        else
            value += (aIn[1] << 16);

        if(aIn[2] < 0)
            value += (256 + aIn[2] << 8);
        else
            value += (aIn[2] << 8);

        if(aIn[3] < 0)
            value += (256 + aIn[3]);
        else
            value += aIn[3];

        return value;
    }

    // issue the download
    public void issueDownload()
    {
        int numDownload = 0;
        synchronized(piecesToDownload)
        {
            while(piecesToDownload.size() < KMaxPieceRequests)
            {
                MTPiece pieceToDownload = torrent.getPieceToDownload(peer);
                if(pieceToDownload != null)
                {
                    piecesToDownload.addElement(new MTPieceToDownload(pieceToDownload, ellapsedTime));
                //iLastRequestTime = iEllapsedTime;
                }
                else
                    break;
            }
            numDownload = piecesToDownload.size();
        }

        if(numDownload == 0)
        {
            setInterested(false);

            if(ellapsedTime > 15)
            {
                if(!isPeerInterested())
                {
                    torrentMgr.notifyTorrentObserverMain(torrent, MTTorrentObserver.EMTMainEventTorrentUploadEnded);

                    int closeOrder = ENotSpecified;
                    if(torrent.isComplete())
                        closeOrder = EDeletePeer;

                    close("No needed piecese and peer not interested");
                }
                else if(incomingRequests.size() == 0) // wait for possible incoming interested message
                {
                    close("No pieces need and peer is not interested");
                }
            }
        }
        else
        {
            setInterested(true);

            if(!isPeerChoking())
            {
                synchronized(piecesToDownload)
                {
                    for(int i = 0; i < piecesToDownload.size(); i++)
                    {
                        if(!((MTPieceToDownload) piecesToDownload.elementAt(i)).hasPendingRequest)
                        {
                            sendRequestMessage(((MTPieceToDownload) piecesToDownload.elementAt(i)));
                            ((MTPieceToDownload) piecesToDownload.elementAt(i)).hasPendingRequest = true;
                        }
                    }
                }
            }
        }
    }

    // issue the upload
    public void issueUpload()
    {
        if(!isChoking())
            while(incomingRequests.size() > 0)
            {
                BlockRequest block = (BlockRequest) incomingRequests.elementAt(0);
                sendPieceMessage(block.pieceIndex, block.begin, block.length);
                incomingRequests.removeElementAt(0);
            }
    }

    /** Creates a new instance of MTPeerConnection */
    public MTPeerConnection(MTPeer aPeer, MTTorrent aTorrent, MTTorrentManager aTorrentMgr)
    {
        peerSender= new ThreadedPeerSender();
        peer = aPeer;
        torrent = aTorrent;
        torrentMgr = aTorrentMgr;
        incomingConnection = false;

        setInterested(false);
        setPeerInterested(false);

        setChoking(true);
        setPeerChoking(true);

        piecesToDownload = new Vector();
        incomingRequests = new Vector();

        state = MTPeerConnection.EPeerNotConnected; ////////////////////////////////////////////////////////
    }

    // initialize via a socket - use when incoming connection
    public void initializeIncomingConnection(SocketConnection aSocket)
    {
        log("InitializeIncomingConnection from StreamConnection begin");
        incomingConnection = true;
        retries++;

        // save socket and open streams
        socket = aSocket;
        try
        {
            inputStream = socket.openInputStream();
            outputStream = socket.openOutputStream();
            peerSender.setOutputStream(outputStream);
        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Opening streams failed - " + ex.getMessage());
            torrentMgr.notifyStatusChanged("Opening streams failed on: " + peer.getAddress() + " | " + ex.getMessage());
            return;
        } catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

        setInterested(false);
        setPeerInterested(false);

        setChoking(true);
        setPeerChoking(true);

        // increment incoming connection
        torrentMgr.incIncomingConnectionCount();

        // change state, because it is already connected
        changeState(EPeerPwHandshaking);

        // start reading
        readEnabled = true;
        log("InitializeIncomingConnection from StreamConnection end");
        read();
    }

    public void onTimer()
    {
        ellapsedTime++;
        if(reconnectAfter > 0)
            reconnectAfter--;

        switch(state)
        {
            case EPeerTcpConnecting:
                {
                    if(ellapsedTime > KTcpConnectTimeout)
                    {
                        tcpConnectionTimeoutNum++;
                        close(EIncreaseErrorCounter, "Timeout while trying to connect");
                    }
                }
                break;

            case EPeerPwHandshaking:
                {
                    if(ellapsedTime > KHandshakeTimeout)
                    {
                        close(EIncreaseErrorCounter, "Handshake timeout (no data received)");
                    }
                }
                break;

            case EPeerPwConnected:
                {
                    if((ellapsedTime - lastMessageReceivedTime) > KPwConnectionTimeout)
                    {
                        close(EIncreaseErrorCounter, "General timeout (no data received)");
                        break;
                    }

                    if(lastRequestTime > 0 && ((ellapsedTime - lastRequestTime) > KRequestTimeout))
                    {
                        lastRequestTime = ellapsedTime;
                        if(torrent.hasTimeoutlessPeer())
                        {
                            peer.setHadRequestTimeout(true);
                            close(EIncreaseErrorCounter, "Request timeout");
                            break;
                        }
                    }

                    if((ellapsedTime > 10) && (!isInterested()) && (!isPeerInterested()))
                    {
                        torrentMgr.notifyTorrentObserverMain(torrent, MTTorrentObserver.EMTMainEventTorrentUploadEnded);
                        close("Nobody interested!");
                    }

                    if((ellapsedTime - lastMessageSentTime) >= KKeepAliveInterval)
                        sendKeepAliveMessage();

                    if(piecesToDownload.size() == 0)
                    {
                        //System.out.println("REREQUEST");
                        issueDownload();
                    }
                }
                break;

            default:
                ;
        }
    }

    private void calculateLongConnection()
    {
        if(MTPeerConnection.tcpConnectionTimeoutNum % 5 == 0)
        {
            MTPeerConnection.useLongConnection = !MTPeerConnection.useLongConnection;
        }
    }

    class ConnectThread extends Thread
    {

        private boolean ok = true;

        public void run()
        {
            try
            {
                retries++;
                reconnectAfter = 0;
                changeState(MTPeerConnection.EPeerTcpConnecting);
                ellapsedTime = 0;

                if(peer.isIpOK() || (peer.isIpCheckTimeout() &&
                    MTNetworkStatusManager.testIP(peer.getAddress(), peer.getPort(), torrent, torrentMgr.getSatistics())
                    ))
                {
                    peer.setIpOK(true);
                    peer.setLastIPCheckTime(System.currentTimeMillis());
                    String address = remoteAddress();
                    try
                    {
                        //String address = "127.0.0.1:10000";
                        torrentMgr.notifyStatusChanged("Trying to connect to: " + address);
                        log("--- Trying to connect to: " + address);

                        if(MTPeerConnection.useLongConnection)
                            socket = (SocketConnection)MTNetworkStatusManager.connect("socket://" + address, Connector.READ_WRITE, true);
                        else
                            socket = (SocketConnection)MTNetworkStatusManager.connect("socket://" + address);

                        if(!ok)
                        {
                            MTLogger.writeLine("--- Connect thread finished after dead: " + address);
                            if(socket != null)
                            {
                                socket.close();
                            }
                            socket = null;
                            return;
                        }

                        readEnabled = true;
                        changeState(MTPeerConnection.EPeerConnected);
                        torrentMgr.notifyStatusChanged("Succesfull connection to: " + address);

                        log("TCP connecting successed");
                        startDownloading();
                    }
                    catch(IOException ex)
                    {
                        close(EIncreaseErrorCounter, "Connecting failed - " + ex.getMessage());
                        torrentMgr.notifyStatusChanged("Connection failed to (IO): " + address + " | " + ex.getMessage());
                        calculateLongConnection();
                    }
                    catch(Exception ex)
                    {
                        close(EIncreaseErrorCounter, "Connecting failed (general)- " + ex.getMessage());
                        torrentMgr.notifyStatusChanged("Connection failed to (general): " + address + " | " + ex.getMessage());
                    }
                }
                else
                {
                    peer.setIpOK(false);
                    peer.setLastIPCheckTime(System.currentTimeMillis());
                    close(EIncreaseErrorCounter, "Connecting failed - IPCheck failed");
                    torrentMgr.notifyStatusChanged("Connection failed - IPCheck error: " +
                        peer.getAddress() + ":" + peer.getPort());
                }
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
        }

        public void setOk(boolean ok)
        {
            this.ok = ok;
        }
    }

    public void connect()
    {
        connectThread = new ConnectThread();
        //connectThread.setPriority(Thread.MAX_PRIORITY);
        connectThread.start();
    }

    public void startDownloading()
    {
        torrentMgr.notifyStatusChanged("Opening streams on: " + peer.getAddress());

        try
        {
            inputStream = socket.openInputStream();
            outputStream = socket.openOutputStream();
            peerSender.setOutputStream(outputStream);

        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Opening streams failed - " + ex.getMessage());
            torrentMgr.notifyStatusChanged("Opening streams failed on: " + peer.getAddress() + " | " + ex.getMessage());
            return;
        } catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

        torrentMgr.notifyStatusChanged("Start download from: " + peer.getAddress());
        changeState(MTPeerConnection.EPeerPwHandshaking);
        sendHandshakeMessage();

        ellapsedTime = 0;
        //System.out.println("STARTREAD: "+peer.getAddress());
        new Thread()
        {

            public void run()
            {
                read();
            }
        }.start();
    }

    public void sendHandshakeMessage()
    {
        if(torrent != null)
        {
            log("Sending handshake");
            try
            {
                //if(outputStream != null)
                //{
                    byte[] zeros =
                    {
                        0, 0, 0, 0, 0, 0, 0, 0
                    };

                    ByteArrayOutputStream bs = new ByteArrayOutputStream();
                    bs.write((byte) ProtocolId.length());
                    bs.write(ProtocolId.getBytes());
                    bs.write(zeros);
                    bs.write(torrent.getInfoHashByteArray());
                    bs.write(torrentMgr.getPeerID().getBytes());

                  

                    int ret=peerSender.queueControlMessage(bs.toByteArray());
                    if(ret==1)
                        close("ERROR, while send handshake, outputstream is NULL");
                  /**  if(ret==1)
                    {
                        log("ERROR, while send handshake");
                        close(EIncreaseErrorCounter, "Error while writing");
                    } **/
                    //outputStream.write(bs.toByteArray());
                    //outputStream.flush();

                    log("Handshake sent");
                //}
               // else
               // {
               //     close("ERROR, while send handshake, outputstream is NULL");
               // }
            }
            catch(IOException ex)
            {
                log("ERROR, while send handshake");
                close(EIncreaseErrorCounter, "Error while writing");
            }
            catch(Exception ex)
            {
                log("ERROR, while send handshake");
                close(EIncreaseErrorCounter, "Error while writing");
            }
        }
        else
            log("ERROR, torrent is not specified, cannot send handshake");
    }

    public String remoteAddress()
    {
        return peer.getAddress() + ":" + peer.getPort();
    }

    public void changeState(int aState)
    {
        state = aState;
        ellapsedTime = 0;
    }

    public void log(String aText)
    {
        MTLogger.writeLine("[Peer " + peer.getAddress() + ":" + peer.getPort() +
            " " + peer.getClient() + "]");
        MTLogger.writeLine("\t" + aText);
    }

    public int state()
    {
        return state;
    }

    private void setPeerWireConnected()
    {
        peerWireConnected = true;
        torrent.increasePWConnectionCount();
    }

    public void close(String aReason)
    {
        close(ENotSpecified, aReason);
    }

    public void close(int aOrder, String aReason)
    {
        log("Closing connection. Reason: " + aReason);
        if(state != EPeerClosing)
        {
            closeOrder = aOrder;

            // stop receiving
            readEnabled = false;

            changeState(MTPeerConnection.EPeerClosing);
            if(torrent != null)
            {
                torrent.peerDisconnected(peer, peerWireConnected);
            }
            peerWireConnected = false;

            if(connectThread != null)
            {
                connectThread.setOk(false);
                if(connectThread.isAlive())
                {
                    connectThread.interrupt();
                }
                connectThread = null;
            }

            try
            {
                if(inputStream != null)
                {
                    inputStream.close();
                    inputStream = null;
                }
                if(outputStream != null)
                {
                    outputStream.close();
                    outputStream = null;
                }
                if(socket != null)
                {
                    socket.close();
                    socket = null;
                }
            }
            catch(Exception ex)
            {
                MTLogger.write("Exception while closing: " + ex.getMessage());
            }
            inputStream = null;
            outputStream = null;
            socket = null;

        /*if (iReconnectAfter <= 0)
        iReconnectAfter = KDefaultDelayBeforeReconnecting;*/
        }
    }

    private boolean isChoking()
    {
        return (KFlagAmChoking & statusFlags) > 0;
    }

    private boolean isPeerChoking()
    {
        return (KFlagPeerChoking & statusFlags) > 0;
    }

    private void setChoking(boolean aChoking)
    {
        if(aChoking)
        {
            if(!isChoking())
            {
                statusFlags |= KFlagAmChoking;

                if(state == EPeerPwConnected)
                {
                    //incomingRequests.Reset();///////////////////////////////////
                    sendChokeMessage();
                }
            }
        }
        else if(isChoking())
        {
            statusFlags &= (~KFlagAmChoking);

            if(state == EPeerPwConnected)
            {
                sendUnchokeMessage();
            //IssueUploadL();////////////////////////////////////////////////
            }
        }
    }

    private void setPeerChoking(boolean aChoking)
    {
        if(aChoking)
            statusFlags |= KFlagPeerChoking;
        else
            statusFlags &= (~KFlagPeerChoking);
    }

    private boolean isInterested()
    {
        return (KFlagAmInterested & statusFlags) > 0;
    }

    private boolean isPeerInterested()
    {
        return (KFlagPeerInterested & statusFlags) > 0;
    }

    private void setInterested(boolean aInterested)
    {
        if(aInterested)
        {
            if(!isInterested())
            {
                statusFlags |= KFlagAmInterested;

                if(state == EPeerPwConnected)
                    sendInterestedMessage();
            }
        }
        else if(isInterested())
        {
            statusFlags &= (~KFlagAmInterested);

            if(state == EPeerPwConnected)
            {
                sendNotInterestedMessage();

                if(!isPeerInterested())
                    close("Nobody interested");
            }

        }
    }

    private void setPeerInterested(boolean aInterested)
    {
        if(aInterested)
            statusFlags |= KFlagPeerInterested;
        else
        {
            statusFlags &= (~KFlagPeerInterested);

            if((state == EPeerPwConnected) && (!isInterested()))
                close("Nobody interested");
        }

    }

    private byte[] putIntToSendBuffer(int aInteger)
    {
        byte[] buffer = new byte[4];

        buffer[3] = (byte) (aInteger & 0xFF);
        buffer[2] = (byte) ((aInteger & (0xFF << 8)) >> 8);
        buffer[1] = (byte) ((aInteger & (0xFF << 16)) >> 16);
        buffer[0] = (byte) ((aInteger & (0xFF << 24)) >> 24);

        return buffer;
    }

    public void sendKeepAliveMessage()
    {
        try
        {
            //if(outputStream != null)
            //{
                lastMessageSentTime = ellapsedTime;
                // if more than this 0 is need to send then use ByteArrayOutputStream
                //outputStream.write(putIntToSendBuffer(0));
                int ret=peerSender.queueControlMessage(putIntToSendBuffer(0));
                if(ret==1)
                    close("ERROR, while send keepalive, outputstream is NULL");
                //outputStream.flush();
               // if(ret==1)
               //     close(EIncreaseErrorCounter, "Error while writing keepalive");
            //}
            //else
            //{
            //    close("ERROR, while send keepalive, outputstream is NULL");
            //}
        }
/*        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Error while writing keepalive");
        }   */
        catch(Exception ex)
        {
            close(EIncreaseErrorCounter, "Error while writing keepalive");
        }
    }

    private void sendBitfieldMessage() /////////////////////////////////////////
    {
        byte[] bitfield = torrent.getBitField().data();
        try
        {
            //if(outputStream != null)
            //{
                lastMessageSentTime = ellapsedTime;

                ByteArrayOutputStream bs = new ByteArrayOutputStream();
                bs.write(putIntToSendBuffer(1 + bitfield.length));
                bs.write(KMessageIdBitfield);
                bs.write(bitfield);
                bs.flush();

                //outputStream.write(bs.toByteArray());
                int ret=peerSender.queueControlMessage(bs.toByteArray());
                if(ret==1)
                    close("ERROR, while send bitfield, outputstream is NULL");
                //outputStream.flush();
                
                bs.close();
                bs = null;

                log("out BITFIELD");
            //}
            //else
            //{
            //    close("ERROR, while send bitfield, outputstream is NULL");
            //}
        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Error while writing bitfield");
        }
        catch(Exception ex)
        {
            close(EIncreaseErrorCounter, "Error while writing bitfield");
        }
    }

    private void sendFakeBitfieldMessage() /////////////////////////////////////////
    {
        byte[] bitfield = new byte[torrent.getBitField().lengthInBytes()];
        for(int i = 0; i < bitfield.length; i++)
            bitfield[i] = (byte) 0;

        try
        {
            //if(outputStream != null)
            //{
                lastMessageSentTime = ellapsedTime;

                ByteArrayOutputStream bs = new ByteArrayOutputStream();
                bs.write(putIntToSendBuffer(1 + bitfield.length));
                bs.write(KMessageIdBitfield);
                bs.write(bitfield);
                bs.flush();

                //outputStream.write(bs.toByteArray());
                 int ret=peerSender.queueControlMessage(bs.toByteArray());
                if(ret==1)
                    close("ERROR, while send bitfield, outputstream is NULL");
               // if(ret==1)
               // close(EIncreaseErrorCounter, "Error while writing bitfield");
                //outputStream.flush();

                bs.close();
                bs = null;

                log("out FAKEBITFIELD");
          //  }
          //  else
          //  {
          //      close("ERROR, while send bitfield, outputstream is NULL");
          //  }
        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Error while writing bitfield");
        }
        catch(Exception ex)
        {
            close(EIncreaseErrorCounter, "Error while writing bitfield");
        }
    }

    private void sendInterestedMessage()
    {
        try
        {
            //if(outputStream != null)
            //{
                lastMessageSentTime = ellapsedTime;

                ByteArrayOutputStream bs = new ByteArrayOutputStream();
                bs.write(putIntToSendBuffer(1));
                bs.write(KMessageIdInterested);
                bs.flush();

                //outputStream.write(bs.toByteArray());
                int ret=peerSender.queueControlMessage(bs.toByteArray());
                if(ret==1)
                    close("ERROR, while send interested, outputstream is NULL");
              //  if(ret==1)
              //      close(EIncreaseErrorCounter, "Error while writing interested: " ); //handle the exception properly here
                //outputStream.flush();

                bs.close();
                bs = null;

                log("out INTERESTED");
           // }
           // else
           // {
           //     close("ERROR, while send interested, outputstream is NULL");
           // }
        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Error while writing interested: " + ex.getMessage());
        }
        catch(Exception ex)
        {
            close(EIncreaseErrorCounter, "Error while writing interested: " + ex.getMessage());
        }
    }

    private void sendNotInterestedMessage()
    {
        try
        {
           // if(outputStream != null)
           // {
                lastMessageSentTime = ellapsedTime;

                ByteArrayOutputStream bs = new ByteArrayOutputStream();
                bs.write(putIntToSendBuffer(1));
                bs.write(KMessageIdNotInterested);
                bs.flush();

                //outputStream.write(bs.toByteArray());
                int ret=peerSender.queueControlMessage(bs.toByteArray());
                if(ret==1)
                    close("ERROR, while send notinterested, outputstream is NULL");
           //         close(EIncreaseErrorCounter, "Error while writing notinterested");
                //outputStream.flush();

                bs.close();
                bs = null;

                torrent.increaseNotInterestedMessages();
                log("out NOTINTERESTED");
         //   }
         //   else
         //   {
         //       close("ERROR, while send notinterested, outputstream is NULL");
         //   }
        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Error while writing notinterested");
        }
        catch(Exception ex)
        {
            close(EIncreaseErrorCounter, "Error while writing notinterested");
        }
    }

    private void sendChokeMessage()
    {
        try
        {
          //  if(outputStream != null)
          //  {
                lastMessageSentTime = ellapsedTime;

                ByteArrayOutputStream bs = new ByteArrayOutputStream();
                bs.write(putIntToSendBuffer(1));
                bs.write(KMessageIdChoke);
                bs.flush();

                //outputStream.write(bs.toByteArray());
               int ret=peerSender.queueControlMessage(bs.toByteArray());
                if(ret==1)
                    close("ERROR, while send notinterested, outputstream is NULL");
                
                //outputStream.flush();

                bs.close();
                bs = null;

                log("out CHOKE");
      //      }
      //     else
      //      {
      //          close("ERROR, while send choke, outputstream is NULL");
      //      }
        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Error while writing choke");
        }
        catch(Exception ex)
        {
            close(EIncreaseErrorCounter, "Error while writing choke");
        }
    }

    private void sendUnchokeMessage()
    {
        try
        {
        //    if(outputStream != null)
        //    {
                lastMessageSentTime = ellapsedTime;

                ByteArrayOutputStream bs = new ByteArrayOutputStream();
                bs.write(putIntToSendBuffer(1));
                bs.write(KMessageIdUnchoke);
                bs.flush();

                //outputStream.write(bs.toByteArray());
                int ret=peerSender.queueControlMessage(bs.toByteArray());
                if(ret==1)
                    close("ERROR, while send notinterested, outputstream is NULL");
                outputStream.flush();

                bs.close();
                bs = null;

                log("out UNCHOKE");
      //      }
      //      else
      //      {
      //          close("ERROR, while send unchoke, outputstream is NULL");
      //      }
        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Error while writing unchoke");
        }
        catch(Exception ex)
        {
            close(EIncreaseErrorCounter, "Error while writing unchoke");
        }
    }

    private void sendRequestMessage(MTPieceToDownload aPiece)
    {
        if(aPiece != null)
        {
            try
            {
              //  if(outputStream != null)
              //  {
                    lastMessageSentTime = ellapsedTime;

                    ByteArrayOutputStream bs = new ByteArrayOutputStream();
                    bs.write(putIntToSendBuffer(13));
                    bs.write(KMessageIdRequest);
                    bs.write(putIntToSendBuffer(aPiece.piece.index()));
                    bs.write(putIntToSendBuffer(aPiece.piece.getDownloadedSize()));
                    int blockLength =
                        aPiece.piece.getTotalSize() - aPiece.piece.getDownloadedSize();
                    if(blockLength > KDefaultBlockLength)
                        blockLength = KDefaultBlockLength;
                    bs.write(putIntToSendBuffer(blockLength));
                    bs.flush();

                    //outputStream.write(bs.toByteArray());
                   int ret=peerSender.queueControlMessage(bs.toByteArray());
                    if(ret==1)
                     close("ERROR, while send notinterested, outputstream is NULL");
                    outputStream.flush();

                    bs.close();
                    bs = null;

                    aPiece.lastRequestLength = blockLength;
                    aPiece.lastRequestBegin = aPiece.piece.getDownloadedSize();

                    log("out REQUEST " + aPiece.piece.index() + "from: " + aPiece.piece.getDownloadedSize() + " (block length: " + blockLength + ")");
             //   }
             //   else
             //   {
             //       close("ERROR, while send request, outputstream is NULL");
             //   }
            }
            catch(IOException ex)
            {
                close(EIncreaseErrorCounter, "Error while writing request");
            }
            catch(Exception ex)
            {
                close(EIncreaseErrorCounter, "Error while writing request");
            }
        }
    }

    public void sendHaveMessage(int aPieceIndex)
    {
        try
        {
          //  if(outputStream != null)
          //  {
                lastMessageSentTime = ellapsedTime;

                ByteArrayOutputStream bs = new ByteArrayOutputStream();
                bs.write(putIntToSendBuffer(5));
                bs.write(KMessageIdHave);
                bs.write(putIntToSendBuffer(aPieceIndex));
                bs.flush();

                //outputStream.write(bs.toByteArray());
               int ret=peerSender.queueControlMessage(bs.toByteArray());
                if(ret==1)
                    close("ERROR, while send notinterested, outputstream is NULL");
                outputStream.flush();

                bs.close();
                bs = null;

                log("out HAVE Piece: " + aPieceIndex);
      //      }
      //      else
      //      {
      //          close("ERROR, while send have, outputstream is NULL");
      //      }
        }
        catch(IOException ex)
        {
        }
        catch(Exception ex)
        {
        }
    }

    private void sendPieceMessage(int aPieceIndex, int aBegin, int aLength)
    {
        /*MTPiece piece = torrent.piece(aPieceIndex);

        if(piece != null)
        {
            log("Processing piece request " + aPieceIndex + " Begin: " + aBegin + " Length: " + aLength + " while piece totalsize: " + piece.getTotalSize());

            if(aBegin + aLength > piece.getTotalSize())
            {
                close("Bad PIECE request (index is out of bounds)");
                return;
            }

            byte[] block = piece.getBlock(aBegin, aLength);

            if(block == null)
            {
                close("Failed to extract block of piece");
                return;
            }  */
            try
            {
               // if(outputStream != null)
               // {
                  /*  ByteArrayOutputStream bs = new ByteArrayOutputStream();
                    bs.write(putIntToSendBuffer(9 + aLength));
                    bs.write(KMessageIdPiece);
                    bs.write(putIntToSendBuffer(aPieceIndex));
                    bs.write(putIntToSendBuffer(aBegin));
                    bs.write(block);
                    bs.flush(); */
                   
                    int ret = peerSender.queueDataMessage(aPieceIndex,aBegin,aLength,torrent, peer); 
                    
                    switch(ret)
                    {
                    case 0:
                	{
                		torrent.updateBytesUploaded(aLength, true);

                        lastMessageSentTime = ellapsedTime;
                        log("out PIECE Index: " + aPieceIndex + " Begin: " + aBegin + " Length: " + aLength);
                        break;
                	}
                   
                    case 1:
                    	{close("Bad PIECE index");
                        return;}
                    case 2:
                    	{close("Bad PIECE request (index is out of bounds)");
                    	return;}
                    case 3:
                    	{close("Failed to extract block of piece");
                    	return;}
                        case 4:
                        {   close("ERROR, while send piece, outputstream is NULL");
                            return;
                        }
                    }
                    //outputStream.write(bs.toByteArray());
                    //peerSender.send(b,outputStream);
           
                    outputStream.flush();

                   /* bs.close();
                    bs = null;  */
	
         //       }
         //       else
         //       {
                   
         //       	close("ERROR, while send piece, outputstream is NULL");
         //       }
            }
            catch(IOException ex)
            {
                close(EIncreaseErrorCounter, "Error while writing piece " + ex.getMessage());
            }
            catch(Exception ex)
            {
                close(EIncreaseErrorCounter, "Error while writing piece(e) " + ex.getMessage());
            }
        }
    	
    

    void sendCancelMessage(MTPieceToDownload aPiece)
    {
        lastMessageSentTime = ellapsedTime;
        try
        {
            ByteArrayOutputStream bs = new ByteArrayOutputStream();

            bs.write(putIntToSendBuffer(13));
            bs.write(KMessageIdCancel);
            bs.write(putIntToSendBuffer(aPiece.piece.index()));
            bs.write(putIntToSendBuffer(aPiece.lastRequestBegin));
            bs.write(putIntToSendBuffer(aPiece.lastRequestLength));
            bs.flush();

            //outputStream.write(bs.toByteArray());
            peerSender.queueControlMessage(bs.toByteArray());
           // if(ret==1)
           //     close(EIncreaseErrorCounter, "Error while writing cancel message");
            //outputStream.flush();

            bs.close();
            bs = null;

            log("out CANCEL[" + aPiece.piece.index() + "] (block length: " + aPiece.lastRequestTime + ")");
        }
        catch(IOException ex)
        {
            close(EIncreaseErrorCounter, "Error while writing cancel message");
        }
        catch(Exception ex)
        {
            close(EIncreaseErrorCounter, "Error while writing cancel message");
        }
    }

    void cancelPieceRequest(MTPiece aPiece)
    {
        if(state == EPeerPwConnected)
        {
            MTPieceToDownload foundPieceToDownload = null;
            synchronized(piecesToDownload)
            {
                for(int i = 0; i < piecesToDownload.size(); i++)
                {
                    MTPieceToDownload tmp = (MTPieceToDownload) piecesToDownload.elementAt(i);
                    if(tmp.piece.equals(aPiece))
                    {
                        foundPieceToDownload = tmp;
                        piecesToDownload.removeElementAt(i);
                        break;
                    }
                }
            }

            if(foundPieceToDownload != null)
            {
                if(foundPieceToDownload.hasPendingRequest)
                {
                    sendCancelMessage(foundPieceToDownload);
                    issueDownload();
                }
            }
        }
    }

    public void cancelAllPieceRequests()
    {
        if(state == EPeerPwConnected)
        {
            MTPieceToDownload [] tmpCopy;
            int cnt = 0;
            synchronized(piecesToDownload)
            {
                tmpCopy = new MTPieceToDownload[piecesToDownload.size()];
                for(int i = 0; i < piecesToDownload.size(); i++)
                {
                    MTPieceToDownload tempPieceToDownload = (MTPieceToDownload) piecesToDownload.elementAt(i);
                    if(tempPieceToDownload.hasPendingRequest)
                        tmpCopy[cnt++] = tempPieceToDownload;
                }
            }
            
            for(int i = 0; i < cnt; i++)
                sendCancelMessage(tmpCopy[i]);
            tmpCopy = null;
        }
    }

    public String getRemoteAddress()
    {
        return peer.getAddress();
    }

    public int getPort()
    {
        return peer.getPort();
    }

    public MTPeer getPeer()
    {
        return peer;
    }

    public MTPeerConnection getPeerConnection()
    {
        return this;
    }

    public int getCloseOrder()
    {
        return closeOrder;
    }

    public Vector getPiecesToDownload()
    {
        return piecesToDownload;
    }

    public void setTorrent(MTTorrent torrent)
    {
        this.torrent = torrent;
    }

    public SocketConnection getSocket()
    {
        return socket;
    }

    public boolean isIncomingConnection()
    {
        return incomingConnection;
    }
}

